/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2025. All rights reserved.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

#ifndef OMNISTREAM_NEXMARKGENERATOR_H
#define OMNISTREAM_NEXMARKGENERATOR_H

#include <exception>
#include <iterator>
#include <stdexcept>
#include <utility>
#include "../model/Event.h"
#include "../model/EventGenerator.h"
#include "../model/AuctionGenerator.h"
#include "../model/PersonGenerator.h"
#include "../model/BidGenerator.h"

class NexmarkGenerator : public std::iterator<std::input_iterator_tag, NexmarkGenerator> {
public:
    // NextEvent nested class.
    class NextEvent {
    public:
        // When, in wallclock time, should this event be emitted?
        const int64_t wallclockTimestamp;
        // When, in event time, should this event be considered to have occured?
        const int64_t eventTimestamp;
        // The event itself.
        std::unique_ptr<Event> event;
        // The minimum of this and all future event timestamps.
        const int64_t watermark;

        NextEvent(int64_t wallclockTimestamp, int64_t eventTimestamp, std::unique_ptr<Event> event, int64_t watermark)
            : wallclockTimestamp(wallclockTimestamp), eventTimestamp(eventTimestamp), event(std::move(event)), watermark(watermark) {}

        bool operator==(const NextEvent &other) const;

        // A hashCode equivalent; implemented as a function.
        int hashCode() const;

        // CompareTo equivalent operator.
        bool operator<(const NextEvent &other) const;
    };

public:
    // Constructor.
    NexmarkGenerator(const GeneratorConfig &config, int64_t eventsCountSoFar, int64_t wallclockBaseTime)
        : random(SplittableRandom(0)),
          config(config),
          eventsCountSoFar(eventsCountSoFar),
          wallclockBaseTime(wallclockBaseTime) {}

    // Create a fresh generator according to config.
    explicit NexmarkGenerator(const GeneratorConfig &config)
        : NexmarkGenerator(config, 0, -1) {}

    // Return a deep copy of this generator.
    NexmarkGenerator copy() const
    {
        // checkNotNull(config) assumed.
        return NexmarkGenerator(config, eventsCountSoFar, wallclockBaseTime);
    }

    // Return the current config for this generator.
    GeneratorConfig getCurrentConfig() const
    {
        return config;
    }

    // Mutate this generator so that it will only generate events up to but not including eventId.
    // Return a config to represent the events this generator will no longer yield.
    GeneratorConfig splitAtEventId(int64_t eventId);

    // Return the next 'event id'. Though events don't have ids we can simulate them to help with bookkeeping.
    int64_t getNextEventId() const
    {
        return config.firstEventId + config.nextAdjustedEventNumber(eventsCountSoFar);
    }

    // Check if there is a next event.
    bool hasNext() const
    {
        return eventsCountSoFar < config.maxEvents;
    }

    // Return the next event.
    NextEvent nextEvent();

    // next() method to mimic Java's Iterator interface.
    NextEvent next()
    {
        return nextEvent();
    }

    // remove() method throws UnsupportedOperationException equivalent.
    void remove()
    {
        throw std::runtime_error("UnsupportedOperationException");
    }

    // Return an estimate of fraction of output consumed.
    double getFractionConsumed() const
    {
        if (config.maxEvents == 0) {
            GErrorLog("max event num is 0");
            throw std::runtime_error("max event num is 0");
        }
        
        return static_cast<double>(eventsCountSoFar) / config.maxEvents;
    }

    // Gets Number of events generated by this generator.
    int64_t getEventsCountSoFar() const
    {
        return eventsCountSoFar;
    }

    // toString() equivalent.
    std::string toString() const { return "";}
private:
    // SplittableRandom equivalent.
    SplittableRandom random;

    // Configuration to generate events against.
    GeneratorConfig config;

    // Number of events generated by this generator.
    int64_t eventsCountSoFar;

    // Wallclock time at which we emitted the first event (ms since epoch). Initially -1.
    int64_t wallclockBaseTime;

    AuctionGenerator auctionGenerator;
    PersonGenerator personGenerator;
    BidGenerator bidGenerator;
};
#endif // OMNISTREAM_NEXMARKGENERATOR_H
